from sqlalchemy import (
    text,
    create_engine,
)
from sqlalchemy.orm import sessionmaker, Session , object_session
import pandas as pd
from typing import List, Dict, Any
from settings import mysql_config

def get_db(db: str='dev', schema: str=''):
    """获取数据库连接"""
    if not mysql_config.get(db):
        raise Exception("can't find reinder mysql config")
    schema = schema if schema else mysql_config[db]['db']
    return create_engine(f"mysql+pymysql://{mysql_config[db]['user']}:{mysql_config[db]['password']}@{mysql_config[db]['host']}:{mysql_config[db]['port']}/{schema}")

def get_session(db: str='dev', schema: str=''):
    """获取数据库会话"""
    engine = get_db(db, schema)
    Session = sessionmaker(bind=engine, class_=SoftDeleteSession)
    session = Session()
    return session

class SoftDeleteSession(Session):
    
    description = "重载软删除方法"

    def delete(self, instance):
        # 获取实例的 is_active 属性
        if hasattr(instance, 'is_active'):
            # 标记为非活动（即软删除）
            instance.is_active = False
            object_session(instance).add(instance)
        else:
            # 如果没有 is_active 字段，则调用父类的 delete 方法
            super(SoftDeleteSession, self).delete(instance)


class MysqlDbUtil():
    def __init__(self, db, schema=''):
        self.engine = self._get_engine_(db)
        self.session = self.get_session()

    @staticmethod
    def _get_engine_(db: str='dev', schema: str=''):
        _config = mysql_config.get(db, {})
        if not _config:
            raise Exception("can't find reinder mysql config")
        schema = schema if schema else _config['db']
        engine = create_engine(f"mysql+pymysql://{_config['user']}:{_config['password']}@{_config['host']}:{_config['port']}/{schema}")
        return engine

    def get_session(self):
        engine = self.engine
        if not self.engine:
            raise Exception("engine is None")
        Session = sessionmaker(bind=engine)
        session = Session()
        return session

    def close(self):
        self.session.close()

    def query(self, sql: str='') -> pd.DataFrame:
        if not sql:
            raise Exception("sql can not be empty")
        res = self.session.execute(text(sql))
        if res.rowcount == 0:
            return pd.DataFrame()
        return pd.DataFrame(res.fetchall(), columns=res.keys())
    
    def query_all(self, sql: str='') -> List[Dict[str, Any]]:
        if not sql:
            raise Exception("sql can not be empty")
        res = self.session.execute(text(sql))
        if res.rowcount == 0:
            return []
        return [dict(row) for row in res.fetchall()]
    
    def insertRecords(self, table: str, schema: str, new_records: Dict[str, Any], commit: bool=True) -> dict:
        if isinstance(new_records, pd.DataFrame):
            new_records = new_records.to_dict(orient='records')
        if isinstance(new_records, dict):
            params = new_records
        if isinstance(new_records, list):
            params = new_records[0]
        if not new_records:
            raise Exception("data to insert is None")
        
        keys = ', '.join(params.keys())
        kk = [':' + key for key in params.keys()]
        values = ', '.join(kk)
        sql = f"INSERT INTO `{schema}`.`{table}` ({keys}) VALUES ({values})"
        try:
            self.session.execute(text(sql), new_records)
            if commit:
                self.session.commit()
            else:
                return {'success': False, 'msg': 'data need to commit'}
            return {'success': True, 'msg': 'success'}
        except Exception as e:
            self.session.rollback()
            self.session.close()
            return {'success': False, 'msg': 'fail to insert ,error:' + str(e)}
        
    def updateRecords(self, table, schema, old_records, new_records) -> dict:
        """update records, obselete old records and insert new records"""
        res = self.deleteRecords(table, schema, old_records, commit=False)
        if not res['success']:
            res['msg'] = 'delete old_records failed'
            return res
        res = self.insertRecords(table, schema, new_records, commit=False)
        if not res['success']:
            res['msg'] = 'insert new_records failed'
            return res
        
        try:
            self.session.commit()
            return {'success': True, 'msg': 'success update'}
        except Exception as e:
            self.session.rollback()
            self.session.close()
            return {'success': False, 'msg': 'fail to update ,error:' + str(e)}
        
    def modifyRecords(self, table, schema, old_records, new_records) -> dict:
        """update records, delete old records and insert new records"""
        if not old_records or not new_records:
            return {'success': False, 'msg': 'old_records or new_records is empty'}
        
        sql = f"UPDATE `{schema}`.`{table}` SET `flag`=1 "
        for k, v in new_records.items():
            if isinstance(v, list) or isinstance(v, dict):
                raise TypeError('value can not be list or dict')
            sql += f", `{k}`='{v}' "
        sql += f"WHERE `flag`=1 "
        for k, v in old_records.items():
            sql += f"AND `{k}`='{v}' "

        try:
            self.session.execute(text(sql))
            self.session.commit()
            return {'success': True, 'msg': 'success update'}
        except Exception as e:
            self.session.rollback()
            self.session.close()
            return {'success': False, 'msg': 'fail to update ,error:' + str(e)} 
        
    def deleteRecords(self, table, schema, old_records, commit=True) -> dict:
        """delete records set flag=0"""
        if isinstance(old_records, pd.DataFrame):
            old_records = old_records.to_dict(orient='records')
        if isinstance(old_records, dict):
            old_records = [old_records]

        for params in old_records:
            res = self.__deleteRecord__(table, schema, params)
            if not res['success']:
                return res
        if commit:
            try:
                self.session.commit()
                return {'success': True, 'msg': 'success delete'}
            except Exception as e:
                self.session.rollback()
                self.session.close()
                return {'success': False, 'msg': 'fail to delete ,error:' + str(e)}
        else:
            return {'success': True, 'msg': 'delete opereation need to commit'}

    def __deleteRecord__(self, table, schema, old_record) -> dict:
        """delete single record set flag=0"""
        if isinstance(old_record, dict):
            params = old_record
        else:
            return {'success': False, 'msg': 'old_record must be dict'}
        
        sql = f"update `{schema}`.`{table}` set flag=0 where flag=1"
        update_key = {}
        for k,v in params.items():
            if isinstance(v, list):
                if isinstance(v[0], int):
                    v = [str(i) for i in v]
                    update_key.update({k: f'({",".join(v)})'})
                else:
                    update_key.update({k: "','".join(v)})
                sql += f" and `{k}` in (:{k})"
                continue
            sql += f" and `{k}`=:{k}"
        params.update(update_key)
        try:
            self.session.execute(text(sql), old_record)
            return {'success': True, 'msg': 'success delete'}
        except Exception as e:
            self.session.rollback()
            self.session.close()
            return {'success': False, 'msg': 'fail to delete ,error:' + str(e)}
        
    def deleteCascadeRecords(self, table, schema, old_records, commit=True) -> dict:
        ...